package com.ruyuan.event.streaming.process;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ruyuan.event.streaming.pojo.EventJoinLog;
import com.ruyuan.event.streaming.pojo.EventLogDTO;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

public class EventJoinReportFlatMap extends RichFlatMapFunction<EventJoinLog,String> {
    ObjectMapper objectMapper;

    @Override
    public void open(Configuration parameters) throws Exception {
        objectMapper = new ObjectMapper();
        super.open(parameters);
    }

    @Override
    public void flatMap(EventJoinLog eventJoinLog, Collector<String> collector) throws Exception {
        EventLogDTO eventLogDTO = ProcessETLUtils.buidEventLogDTO(eventJoinLog);
        collector.collect(objectMapper.writeValueAsString(eventLogDTO));
    }
}